package com.raft.kvstore.server;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.Serializable;
import java.util.List;

import com.alibaba.com.caucho.hessian.io.HessianInput;
import com.raft.core.service.AddNodeCommand;
import com.raft.core.service.RemoveNodeCommand;
import com.raft.kvstore.MessageConstants;
import com.raft.kvstore.message.Failure;
import com.raft.kvstore.message.GetCommand;
import com.raft.kvstore.message.GetCommandResponse;
import com.raft.kvstore.message.Redirect;
import com.raft.kvstore.message.SetCommand;
import com.raft.kvstore.message.Success;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

/**按照消息类型-长度-内容
 * @author zhangsulei
 *
 */
public class Decoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    	// 等待数据
        if (in.readableBytes() < 8) return;

        in.markReaderIndex();
        int messageType = in.readInt();
        int payloadLength = in.readInt();
        if (in.readableBytes() < payloadLength) {
            in.resetReaderIndex();
            return;
        }

        // 单条数据就绪
        byte[] payload = new byte[payloadLength];
        in.readBytes(payload);
        switch (messageType) {
            case MessageConstants.MSG_TYPE_SUCCESS:
            	// 成功
                out.add(Success.INSTANCE);
                break;
            case MessageConstants.MSG_TYPE_FAILURE:
            	// 失败
                out.add(decode(payload, Failure.class));
                break;
            case MessageConstants.MSG_TYPE_REDIRECT:
            	// 重定向
                out.add(decode(payload, Redirect.class));
                break;
            case MessageConstants.MSG_TYPE_ADD_SERVER_COMMAND:
                out.add(decode(payload, AddNodeCommand.class));
                break;
            case MessageConstants.MSG_TYPE_REMOVE_SERVER_COMMAND:
                out.add(decode(payload, RemoveNodeCommand.class));
                break;
            case MessageConstants.MSG_TYPE_GET_COMMAND:
            	// get命令
                out.add(decode(payload, GetCommand.class));
                break;
            case MessageConstants.MSG_TYPE_GET_COMMAND_RESPONSE:
            	// get命令响应
                out.add(decode(payload, GetCommandResponse.class));
                break;
            case MessageConstants.MSG_TYPE_SET_COMMAND:
            	// set命令
                out.add(decode(payload, SetCommand.class));
                break;
            default:
                throw new IllegalStateException("unexpected message type " + messageType);
        }
    }

    private <T extends Serializable> T decode(byte[] bytes, Class<T> clazz) {
        if (bytes == null) {
            return null;
        }
        // 1、将字节数组转换成字节输入流
        ByteArrayInputStream bis = new ByteArrayInputStream(bytes);

        // step 1. 定义外部序列化工厂
        //ExtSerializerFactory extSerializerFactory = new ExtSerializerFactory();
        //extSerializerFactory.addSerializer(java.time.OffsetDateTime.class, new OffsetDateTimeRedisSerializer());
        //extSerializerFactory.addDeserializer(java.time.OffsetDateTime.class, new OffsetDateTimeRedisDeserializer());
        // step 2. 序列化工厂
        //SerializerFactory serializerFactory = new SerializerFactory();
        //serializerFactory.addFactory(extSerializerFactory);
        HessianInput hessianInput = new HessianInput(bis);
        //hessianInput.setSerializerFactory(serializerFactory);
        Object object = null;

        try {
            object = hessianInput.readObject();
        } catch (IOException e) {
        	e.printStackTrace();
        } catch (Exception e) {
        	e.printStackTrace();
        }

        return (T) object;
    }

}
